package site.wuao.simplerxjava.operator;

import android.util.Log;

import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.Subscriber;
import rx.functions.Func1;

/**
 * 操作符 - 错误
 *
 * @author wuao
 * @date 2017.12.07
 * @note -
 * ---------------------------------------------------------------------------------------------------------------------
 * @modified -
 * @date -
 * @note -
 */
public class ErrorOperator {
    /** 标记 */
    private static final String TAG = "ErrorOperator";

    /**
     * catch_onErrorReturn
     *
     * @note 1.原有Observable遇到错误时返回原有Observable行为的备用Observable
     * 2.备用Observable会忽略原有Observable的onError调用
     * 3.会发射一个特殊的项并且调用观察者的onCompleted方法
     */
    public void onErrorReturn() {
        Observable
                .create(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        for (int i = 0; i < 10; i++) {
                            if (i > 5) {
                                subscriber.onError(new RuntimeException("this is custom exception"));
                            }
                            subscriber.onNext(i);
                        }
                        subscriber.onCompleted();
                    }
                })
                .onErrorReturn(new Func1<Throwable, Integer>() {
                    @Override
                    public Integer call(Throwable throwable) {
                        return 11;
                    }
                })
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.i(TAG, "onCompleted");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "onError: " + e.getMessage());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "onNext: " + integer);
                    }
                });

        // onNext: 0
        // onNext: 1
        // onNext: 2
        // onNext: 3
        // onNext: 4
        // onNext: 5
        // onNext: 11
        // onCompleted
    }

    /**
     * catch_onErrorResumeNext
     *
     * @note 1.原有Observable遇到错误时返回原有Observable行为的备用Observable
     * 2.备用Observable会忽略原有Observable的onError调用
     * 3.会发射一个备用Observable数据并且调用观察者的onCompleted方法
     */
    public void onErrorResumeNext() {
        Observable
                .create(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        for (int i = 0; i < 10; i++) {
                            if (i > 5) {
                                subscriber.onError(new RuntimeException("this is custom exception"));
                            }
                            subscriber.onNext(i);
                        }
                        subscriber.onCompleted();
                    }
                })
                .onErrorResumeNext(new Func1<Throwable, Observable<? extends Integer>>() {
                    @Override
                    public Observable<? extends Integer> call(Throwable throwable) {
                        return Observable.just(11,12,13);
                    }
                })
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.i(TAG, "onCompleted");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "onError: " + e.getMessage());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "onNext: " + integer);
                    }
                });

        // onNext: 0
        // onNext: 1
        // onNext: 2
        // onNext: 3
        // onNext: 4
        // onNext: 5
        // onNext: 11
        // onNext: 12
        // onNext: 13
        // onCompleted
    }

    /**
     * catch_onExceptionResumeNext
     *
     * @note 1.原有Observable遇到错误时返回原有Observable行为的备用Observable
     * 2.备用Observable会忽略原有Observable的onError调用
     * 3.会发射一个备用Observable数据并且调用观察者的onCompleted方法
     * 4.如果onError收到的Throwable不是一个Exception，会将错误传递给观察者的onError方法，不会调用备用的Observable
     */
    public void onExceptionResumeNext() {
        Observable
                .create(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        for (int i = 0; i < 10; i++) {
                            if (i > 5) {
                                subscriber.onError(new Error("this is custom error"));
                            }
                            subscriber.onNext(i);
                        }
                        subscriber.onCompleted();
                    }
                })
                .onExceptionResumeNext(Observable.just(11,12,13))
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.i(TAG, "onCompleted");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "onError: " + e.getMessage());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "onNext: " + integer);
                    }
                });

        // onNext: 0
        // onNext: 1
        // onNext: 2
        // onNext: 3
        // onNext: 4
        // onNext: 5
        // onError: this is custom error
    }

    /**
     * retry - 重试
     *
     * @note 1.不会将原始的Observable的onError通知给观察者，会再次执行n次
     * 2.如果次数超过了，会把最新的一个onError通知给观察者
     */
    public void retry() {
        Observable
                .create(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        for (int i = 0; i < 5; i++) {
                            if (i == 1) {
                                subscriber.onError(new RuntimeException("this is custom exception"));
                            } else {
                                subscriber.onNext(i);
                            }
                        }
                        subscriber.onCompleted();
                    }
                })
                .retry(3)
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.i(TAG, "onCompleted");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "onError: " + e.getMessage());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "onNext: " + integer + "  " + System.currentTimeMillis());
                    }
                });

        // onNext: 0  1512636381027
        // onNext: 0  1512636381028
        // onNext: 0  1512636381028
        // onNext: 0  1512636381028
        // onError: this is custom exception
    }

    /**
     * retry - 重试
     *
     * @note 1.不会将原始的Observable的onError通知给观察者，会再次执行n次
     * 2.如果次数超过了，会把最新的一个onError通知给观察者
     */
    public void retryWhen() {
        Observable
                .create(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        for (int i = 0; i < 5; i++) {
                            if (i == 1) {
                                subscriber.onError(new RuntimeException("this is custom exception"));
                            } else {
                                subscriber.onNext(i);
                            }
                        }
                        subscriber.onCompleted();
                    }
                })
                .retryWhen(new RetryWithDelay(3,3000))
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.i(TAG, "onCompleted");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "onError: " + e.getMessage());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "onNext: " + integer + "  " + System.currentTimeMillis());
                    }
                });

        // onNext: 0  1512636309783
        // onNext: 0  1512636312789
        // onNext: 0  1512636315790
        // onNext: 0  1512636318790
        // onError: this is custom exception
    }

    /**
     * 延迟重试
     */
    public static class RetryWithDelay implements
            Func1<Observable<? extends Throwable>, Observable<?>> {
        /** 最大重试次数 */
        private final int mMaxRetries;
        /** 重试延迟(毫秒) */
        private final int mRetryDelay;
        /** 已重试次数 */
        private int mRetryCount;

        /**
         * 构造函数
         *
         * @param maxRetries 最大重试次数
         * @param retryDelay 重试延迟(毫秒)
         */
        public RetryWithDelay(int maxRetries, int retryDelay) {
            this.mMaxRetries = maxRetries;
            this.mRetryDelay = retryDelay;
        }

        @Override
        public Observable<?> call(Observable<? extends Throwable> attempts) {
            return attempts
                    .flatMap(new Func1<Throwable, Observable<?>>() {
                        @Override
                        public Observable<?> call(Throwable throwable) {
                            if (++mRetryCount <= mMaxRetries) {
                                return Observable.timer(mRetryDelay,TimeUnit.MILLISECONDS);
                            }
                            // 超过重试次数，返回错误
                            return Observable.error(throwable);
                        }
                    });
        }
    }
}
